	// Copyright (c) 2007 Intel Corp. (c) 2009 Daniele Buono
//

// Black-Scholes
// Analytical method for calculating European Options
//
//
// Reference Source: Options, Futures, and Other Derivatives, 3rd Edition, Prentice
// Hall, John C. Hull,


#include "./cycle.h"
#include "../../libLC_unif/channel.h"

#include "blkschlsfunc.h"

#include <stdio.h>
#include <stdlib.h>
#include <sys/ipc.h>	// ftok
#include <unistd.h>		// fork
#include <string.h>		// memcpy
#include <sys/wait.h>	// waitpid

#ifdef AFFINITY
#include <sched.h>
#endif

#define MAX_WORKERS 128

fptype* prices;
int numOptions;
int nWorkers;
int nRuns;

void emitter(OptionData* data, int keys[], int elements[]){
	int i,j,k;
	fptype * sptprice;
	fptype * strike;
	fptype * rate;
	fptype * volatility;
	fptype * otime;
	int * otype;

	Channel *out = malloc(sizeof(Channel)*nWorkers);

	// The first is the bigger partition
	void *buffer = malloc((sizeof(fptype) * 5 * elements[0]) + (sizeof(int) * elements[0]));

	// Prepare Channels
	for (i = 0; i < nWorkers; i++) {
		out[i] = LCattach(keys[i]);
	}

	// Start packing and sending
	for (j = 0; j < nRuns; j++) {
		int counter = 0;
		for(i=0; i<nWorkers; i++){
			sptprice = (fptype *) buffer;
			strike = sptprice + elements[i];
			rate = strike + elements[i];
			volatility = rate + elements[i];
			otime = volatility + elements[i];
			otype = (int *) (otime + elements[i]);
			// Prepare data for ith worker
			for(k=0; k<elements[i]; k++){
				otype[k] = (data[counter].OptionType == 'P') ? 1 : 0;
				sptprice[k] = data[counter].s;
				strike[k] = data[counter].strike;
				rate[k] = data[counter].r;
				volatility[k] = data[counter].v;
				otime[k] = data[counter].t;
				counter++;
			}
			// Send data
			printf("Emitter[%u]: invio i dati a : %d",getpid(),i);
			LCsend(out[i],buffer);

			printf("Emitter: dati inviati : %d, valore |%c|\n",i,*(char *)(buffer+5));

		}
	}
	// Send an EOF pattern
	for(i=0; i<nWorkers; i++){
		strcpy((char*)buffer,"EOF");
		// Send data
        printf("Emitter: invio EOF a : %d\n",i);
		LCsend(out[i],buffer);
		printf("....EOF\n");
	}

	// Detach Channels
	for (i = 0; i < nWorkers; i++) {
		LCdetatch(out[i]);
	}
}

void collector(key_t keys[], key_t resultkey, int elements[]){
	int i;
	fptype *data = malloc(sizeof(fptype)*numOptions);
	fptype ** buffers = malloc(sizeof(void*)*nWorkers);
	Channel *in = malloc(sizeof(Channel)*nWorkers);
	Channel result;

	// Prepare Channels
	result = LCattach(resultkey);
	for (i = 0; i < nWorkers; i++) {
		in[i] = LCattach(keys[i+nWorkers]);
	}

	for(i=0; i<nWorkers; i++){
		buffers[i] = (fptype*) LCreceive(in[i]);
	}
	// Start receiving from workers
	while(strncmp((char*)buffers[0],"EOF",4)!=0){
		for(i=0; i<nWorkers; i++){
		    printf("Collector[%u]: ricevo i dati da: %d\n",getpid(),i);
			buffers[i] = (fptype*) LCreceive(in[i]);
			printf("collector: dati ricevuti");
		}
	}

	// Pack and send back the last received message to the parent process
	int counter = 0;
	for(i=0; i<nWorkers; i++){
		memcpy(data+counter,buffers[i],sizeof(fptype)*elements[i]);
		counter += elements[i];
	}
	LCsend(result,(char*)data);

	// Detach Channels
	for (i = 0; i < nWorkers; i++) {
		LCdestroy(in[i]);
	}
	LCdetatch(result);

	exit(0);
}

int worker(int id, key_t inkey, key_t outkey, int elements) {

	int i, k;
	fptype price[NCO];
//	printf("My id is %d. Channel in key: %d. Channel out key: %d Elements: %d\n", id, inkey, outkey, elements);

#ifdef AFFINITY
	cpu_set_t mask;
	/* CPU_ZERO initializes all the bits in the mask to zero. */
	CPU_ZERO(&mask);
	/* CPU_SET sets only the bit corresponding to cpu. */
	CPU_SET(id, &mask);
	/* sched_setaffinity returns 0 in success */
	if (sched_setaffinity(0, sizeof(mask), &mask) == -1)
	{
		printf("WARNING: Could not set CPU Affinity, continuing...\n");
	}
#endif
	int    * otype;
	fptype * sptprice;
	fptype * strike;
	fptype * rate;
	fptype * volatility;
	fptype * otime;

	void* data = NULL;
	fptype *prices = malloc (sizeof(fptype)* elements);

	// Attach to the channel
	Channel in = LCattach(inkey);
	Channel out = LCattach(outkey);
    printf("Worker: %d, pid: %u, ind_base ch:%p, chiave:%d\n",id,getpid(),in,inkey);
	// Receive the first message
	data = LCreceive(in);
	while(strncmp(data,"EOF",4)!=0){
	    printf("Worker %d, dati ricevuti ---- ricevuto |%c|",id,*(char *)(data+5));
		sptprice = (fptype *) data;
		strike = sptprice + elements;
		rate = strike + elements;
		volatility = rate + elements;
		otime = volatility + elements;
		otype = (int *) (otime + elements);

		for(i=0;i<elements;i += NCO){
			// Calling main function to calculate option value based on Black & Scholes's
			// equation.
			BlkSchlsEqEuroNoDiv(price, NCO, &(sptprice[i]), &(strike[i]),
					&(rate[i]), &(volatility[i]), &(otime[i]), &(otype[i]), 0);
			for (k=0; k<NCO; k++) {
				prices[i+k] = price[k];
			}
		}
        printf("Worker [%d,,%u] invio dati\n",id,getpid());
		LCsend(out, (char*) prices);
    printf("Worker %d, dati inviati\n",id);
		// Receive the next one
		printf("Worker [%d,%u], ricevo dati\n",id,getpid());
		data = LCreceive(in);
	}
	// We know for sure that data is longer then prices, so no harm in doing this
	LCsend(out, (char*) data);

	LCdestroy(in);
	LCdetatch(out);

	printf("Worker %d. Closing...\n", id);

	exit(0);
}

int main(int argc, char **argv) {
	int i;

	OptionData* data;
	fptype *results;

	printf("BlackScholes Benchmark - Messages Version\n");
	fflush(NULL);

	if (argc != 5) {
		printf("Usage:\n\t%s <nworkers> <nruns> <inputFile> <outputFile>\n",
				argv[0]);
		exit(1);
	}

	nWorkers = atoi(argv[1]);
	nRuns = atoi(argv[2]);
	char *inputFile = argv[3];
	char *outputFile = argv[4];

	printf("This cpu have a clockrate of %f Mhz\n", (double) CLOCKRATE / 1000000.0);

	ticks clocks;

	data = loadData(inputFile,&numOptions,&nWorkers);

	printf("Num of Options: %d\n", numOptions);
	printf("Num of Runs: %d\n", nRuns);

	printf("Size of data: %ld\n", numOptions * (sizeof(OptionData)
			+ sizeof(int)));

	// Initialization...

	// Define partition sizes
	int elemperworker = (numOptions / nWorkers);
	while(elemperworker%NCO != 0 ){
		elemperworker++;
	}

	Channel tmpchan;
	key_t *keys = malloc(sizeof(key_t) * nWorkers * 2 );
	int	*elements = malloc(sizeof(int) * nWorkers);

	/*
	 * I create the channels here so they exists when workers and emitter starts
	 * However they are immediately detached so they are not retained after forks
	 */
	for (i = 0; i < nWorkers; i++) {
		elements[i] = elemperworker;
		if(i==nWorkers-1)
			elements[i] = numOptions - ((nWorkers-1) * elemperworker);

		keys[i] = ftok(argv[0],i);
		keys[i+nWorkers] = ftok(argv[0],i+nWorkers);
		if(keys[i]==-1){
			printf("Error creating a new key with parameters %s %d\n",argv[0],i);
			exit(-1);
		}
		printf("Chiave [%d] ",keys[i]);
		tmpchan =  LCcreate_as( keys[i], (sizeof(fptype) * 5 * elements[i]) + (sizeof(int) * elements[i]) , 1,1 );
		LCdetatch(tmpchan);
		printf("Chiave [%d] ",keys[i+nWorkers]);
		tmpchan =  LCcreate_as( keys[i+nWorkers], sizeof(fptype) * elements[i], 1,1 );
		LCdetatch(tmpchan);
	}
	key_t resultkey = ftok(argv[0],2*nWorkers);
	tmpchan = LCcreate_ss( resultkey, sizeof(fptype) * numOptions, 1 );
	LCdetatch(tmpchan);

	int pids[MAX_WORKERS+1];
	for (i = 0; i < nWorkers; i++) {
		pid_t pid = fork();
		if(pid==0){ // Child
			worker(i,keys[i], keys[i+nWorkers],elements[i]);
		}
		else if (pid < 0) { // failed to fork
			printf("Failed to fork\n");
			exit(1);
			// Throw exception
		} else{
			pids[i] = pid;
		}
	}
	// Yet another fork, for the collector
	{
		pid_t pid = fork();
		if(pid==0){ // Child
			collector(keys,resultkey,elements);
		}
		else if (pid < 0) { // failed to fork
			printf("Failed to fork\n");
			exit(1);
			// Throw exception
		} else{
			pids[i] = pid;
		}
	}
	Channel resultc = LCattach(resultkey);

	// Let's start the computation!
	clocks = getticks();

	// Act as an emitter
	emitter(data,keys,elements);

	results = (fptype*) LCreceive(resultc);
	// The computation is done!

	clocks = getticks() - clocks;

	free(elements);
	free(keys);

	// Wait for the threads to quit
	for (i = 0; i < nWorkers; i++) {
		int childExitStatus;
		waitpid( pids[i] , &childExitStatus, 0);
	}

	printf("Total computation time: %lf\n", ((double)clocks)/CLOCKRATE);

	printf("Service time: %f\n", ((double)clocks)/ CLOCKRATE / nRuns);

	// Save on times.txt
	FILE *times = fopen("times.txt","a");
	if (times == NULL) {
		printf("ERROR: Unable to open file %s.\n", "times.txt");
		exit(1);
	}
	fprintf(times,"%d\t\t%f\n",nWorkers,((double)clocks)/CLOCKRATE );
	fclose(times);

	printf("Starting saveData\n");
	fflush(stdout);
	saveData(results,outputFile,numOptions);

	free(data);
	LCdestroy(resultc);

	return 0;
}
